
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

import junit.framework.TestCase;

public class HbaseSearch extends TestCase {

	Configuration conf = HBaseConfiguration.create();
	static String tableName = "bingmurls";

	@Override
	protected void setUp() throws Exception {
		// TODO Auto-generated method stub
		super.setUp();
		conf.set("hbase.zookeeper.quorum", "slave01.secsmarts.com,master.secsmarts.com,slave02.secsmarts.com");
		conf.set("zookeeper.znode.parent", "/hbase-unsecure");
	}

	public void testGet() throws IOException {

		selectByRowKey(tableName, "rc90-kaskus.blogspot.com/");

	}

	public void testMultiGet() {
		String fileName = "Bing_mUrls_2017_01_21_1.tsv";
		File file = new File(fileName);
		BufferedReader reader = null;
		try {

			reader = new BufferedReader(new FileReader(file));
			String tempString = null;
			int line = 1;
			long start = System.currentTimeMillis();
			long end = System.currentTimeMillis();
			while ((tempString = reader.readLine()) != null) {
				String url = tempString.split("\t")[1];
				selectByRowKey(tableName, url,false);
				line++;
				if (line % 1000 == 0) {
					end = System.currentTimeMillis();
					System.out.println(String.format("%d lines,time:%d,url:%s", line, (end - start), url));
				}
			}
			end = System.currentTimeMillis();
			System.out.println(String.format("%d lines,time:%d", line, (end - start)));
			reader.close();
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			if (reader != null) {
				try {
					reader.close();
				} catch (IOException e1) {
				}
			}
		}
	}
	public void selectByRowKey(String tablename, String rowKey) throws IOException {
		selectByRowKey(tablename, rowKey, true);
	}
	public void selectByRowKey(String tablename, String rowKey, boolean print) throws IOException {
		HTable table = new HTable(conf, tablename);
		Get g = new Get(Bytes.toBytes(rowKey));
		Result r = table.get(g);
		for (Cell kv : r.rawCells()) {
			String column = new String(kv.getQualifier());
			String value = new String(kv.getValue());
			if (print) {
				System.out.println("column: " + column);
				System.out.println("value: " + value);
			}
		}
	}
}
